package com.eshipenet.shipnet.dp_unpacker

import java.lang

import org.apache.flink.streaming.connectors.kafka.KafkaSerializationSchema
import org.apache.kafka.clients.producer.ProducerRecord
import org.json4s.{DefaultFormats, Formats}
import org.json4s.native.Serialization.write

object EdgeChannelDataSerializer {

  implicit lazy val formats: Formats = DefaultFormats

}

class EdgeChannelDataSerializer
    extends KafkaSerializationSchema[EdgeChannelData] {

  import EdgeChannelDataSerializer._

  override def serialize(
      element: EdgeChannelData,
      timestamp: lang.Long): ProducerRecord[Array[Byte], Array[Byte]] = {
    val topic = "EdgeChannelData2"
    val key = element.id.getBytes
    val obj = element.aux.values ++ Map(
      "id" -> element.id
    )
    val value = write(obj).getBytes
    new ProducerRecord(topic, key, value)
  }

}
